package com.zty.services.impl;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zty.BO.CompleteChunkFileBO;
import com.zty.BO.FileChunkBO;
import com.zty.BO.FileChunkMeta;
import com.zty.BO.MetaFile;
import com.zty.DTO.FileMeta;
import com.zty.VO.BucketVO;
import com.zty.VO.FileChunkMetaVO;
import com.zty.VO.MetaFileVO;
import com.zty.config.ClientConfig;
import com.zty.errors.BusinessException;
import com.zty.errors.EnumClientException;
import com.zty.response.CommonResponse;
import com.zty.services.FileService;
import com.zty.util.Md5Util;
import com.zty.utils.ChunkAddressStrategy;
import com.zty.utils.ChunkDownloaderStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.io.InputStream;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @Author ZTY
 * @Description
 * @Date 11:48 2023-12-25
 **/

@Service
@Slf4j
public class FileServiceImpl implements FileService {

    private final RestTemplate restTemplate;
    private final ClientConfig clientConfig;
    private final ObjectMapper mapper;
    private final ChunkAddressStrategy chunkAddressStrategy;
    private final ChunkDownloaderStrategy chunkDownloaderStrategy;

    public FileServiceImpl(RestTemplate restTemplate,
                           ClientConfig clientConfig, ObjectMapper mapper,
                           ChunkAddressStrategy chunkAddressStrategy,
                           ChunkDownloaderStrategy chunkDownloaderStrategy) {
        this.restTemplate = restTemplate;
        this.clientConfig = clientConfig;
        this.mapper = mapper;
        this.chunkAddressStrategy = chunkAddressStrategy;
        this.chunkDownloaderStrategy = chunkDownloaderStrategy;
    }


    @Override
    public MetaFile getMeta(String bucketName, String filename) {

        //  meta信息自然要从meta服务器中获取

        //  先将filename的后缀名去一下
        if (filename.contains(".")){
            filename = filename.split("\\.")[0];
        }

        String url = clientConfig.getMetaServerAddress() + "/meta/info?bucketName={bucketName}&filename={filename}";
        Map<String, Object> params = new HashMap<>();
        params.put("bucketName",bucketName);
        params.put("filename",filename);

        //  发送请求
        Object response = restTemplate.getForObject(
                url,
                Object.class,
                params
        );

        CommonResponse<MetaFile> commonResponse = mapper.convertValue(response, new TypeReference<CommonResponse<MetaFile>>() {
        });

        //  取出返回的数据，即meta信息
        return commonResponse.getData();
    }

    @Override
    public byte[] downloadChunk(FileChunkMeta chunk) {

        return chunkDownloaderStrategy.download(chunk);
    }

    @Override
    public MetaFileVO meta(FileMeta fileMeta) {

        //  通过http调用meta服务器获取meta信息
        String url = clientConfig.getMetaServerAddress() + "/meta/generate";
        Object response = restTemplate.postForObject(
                url,
                fileMeta,
                Object.class
        );

        //  将返回的结果做一个类型转化
        CommonResponse<MetaFile> commonResponse = mapper.convertValue(response,
                new TypeReference<CommonResponse<MetaFile>>() {
        });

        MetaFile metaFile = commonResponse.getData();

        return buildMetaFileVO(metaFile);

    }

    @Override
    public String upload(String bucketName, MultipartFile file) {
        //  先将文件相关的信息取出
        String originalFilename = file.getOriginalFilename();
        String extension = "";
        if (Objects.nonNull(originalFilename)){
            int dotIndex = originalFilename.lastIndexOf(".");

            //  不是-1才能证明该文件名存在"."
            if (dotIndex != -1){
                //  取出文件后缀名
                extension = originalFilename.substring(dotIndex);
            }
        }

        //  拿到文件源信息 要通过文件源信息去Meta服务器上生成对应的元信息
        FileMeta fileMeta = new FileMeta()
                .setBucketName(bucketName)
                .setExtension(extension)
                .setFilesize(file.getSize());
        String metaServerAddress = clientConfig.getMetaServerAddress();
        //  使用RestTemplate调用Meta接口，返回response
        Object response = restTemplate.postForObject(clientConfig.getMetaServerAddress() + "/meta/generate",
                fileMeta, Object.class);



        //  类型转换，将response中返回的元数据转变为MetaFile类
        CommonResponse<MetaFile> commonResponse = mapper
                .convertValue(response, new TypeReference<CommonResponse<MetaFile>>() {
                });

        //  commonResponse为空表示返回失败
        if (Objects.isNull(commonResponse)){
            throw new BusinessException(EnumClientException.FAILED_TO_GET_META_FILE);
        }

        //  拿到元数据文件
        MetaFile metaFile = commonResponse.getData();

        if (Objects.isNull(metaFile)){
            throw new BusinessException("metaFile为空",EnumClientException.FAILED_TO_GET_META_FILE);
        }

        try {
            //  将每个分片冗余上传到服务器中
            uploadChunks(file,metaFile);
        } catch (Exception e) {
            throw new BusinessException(EnumClientException.FAILED_TO_UPLOAD_CHUNK_FILE);
        }

        return "%s/%s.%s".formatted(bucketName,metaFile.getFilename(),metaFile.getExtension());
    }

    @Override
    public String uploadChunk(String bucketName,
                              String filename,
                              String md5,
                              Integer chunkNo,
                              MultipartFile file) {


        //  拿到该分片的meta信息
//          MetaFile metaFile = getMeta(bucketName, filename);

        //  可以拿到该文件分片的所有副本的详细信息
//          List<FileChunkMeta> chunks = metaFile.getChunks();

        String metaServerAddress = clientConfig.getMetaServerAddress();

        String url = metaServerAddress + "/meta/chunk/info?bucketName={bucketName}&filename={filename}&chunkNo={chunkNo}";
        Map<String,Object> params = new HashMap<>();
        params.put("filename",filename);
        params.put("bucketName",bucketName);
        params.put("chunkNo",chunkNo);

        //  新补充的获取分片信息的方式
        Object resp = restTemplate.getForObject(
                url,
                Object.class,
                params
        );
        CommonResponse<List<FileChunkMeta>> chunkInfoResp = mapper.convertValue(resp,
                new TypeReference<CommonResponse<List<FileChunkMeta>>>() {
        });
        List<FileChunkMeta> chunks = chunkInfoResp.getData();

        //  先判断该分片内容是否上传完整了，即计算MD5
        String realMd5 = Md5Util.getMd5(file);

        //  真实的md5和某个分片的md5不一致，抛出异常
        if (!Objects.equals(md5,realMd5)){
            throw new BusinessException(EnumClientException.CHUNK_FILE_INCOMPLETE);
        }

        //  由于拿到的是所有副本的详细信息，所以每个副本的chunkNo都应该一致，做一下chunkNo一致筛选
        chunks.stream().filter(c -> c.getChunkNo().equals(chunkNo)).collect(Collectors.toList());

        //  循环遍历进行上传
        chunks.forEach(c -> {

            int chunkSize = c.getChunkSize();
            byte[] buffer = new byte[chunkSize];
            try(InputStream inputStream = file.getInputStream()) {

                inputStream.read(buffer);

                FileChunkBO fileChunkBO = new FileChunkBO();

                fileChunkBO.setFilename(c.getFilename())
                        .setChunkNo(c.getChunkNo())
                        .setExtension(c.getExtension())
                        .setChunkSize(c.getChunkSize())
                        .setBucketName(c.getBucketName())
                        .setBytes(buffer);

                //  获取到每个分片所存在在的地址
                String address = chunkAddressStrategy.get(c);

                //  开始上传
                Object response = restTemplate.postForObject(
                        address + "/file/write",
                        fileChunkBO,
                        Object.class);

                if (Objects.isNull(response)){
                    throw new RuntimeException(MessageFormat.format("第 {} 个分片上传失败",c.getChunkNo()));
                }

                CommonResponse<String> commonResponse = mapper.convertValue(response, new TypeReference<CommonResponse<String>>() {
                });

                String serverMd5 = commonResponse.getData();

                //  不一致表示分片不完整
                if (!Objects.equals(serverMd5,realMd5)){
                    throw new RuntimeException(MessageFormat.format("第 {} 个分片不完整",c.getChunkNo()));
                }

                CompleteChunkFileBO completeChunkFileBO = new CompleteChunkFileBO();
                completeChunkFileBO.setFilename(c.getFilename())
                        .setChunkNo(c.getChunkNo())
                        .setSchema(c.getSchema())
                        .setMd5(md5)
                        .setAddress(c.getAddress());

                //  上传成功
                //  将上传成功的消息发至meta服务器
                Object completeResp = restTemplate.postForObject(
                        metaServerAddress + "/meta/chunk/complete",
                        completeChunkFileBO,
                        Object.class
                );

                //  resp为空表示状态更新失败
                if (Objects.isNull(completeResp)){
                    throw new RuntimeException(MessageFormat.format("第 {} 个状态更新失败",c.getChunkNo()));
                }
            } catch (IOException e) {
                log.info("第 {} 个分片上传失败，原因是：",c.getChunkNo(),e);
                throw new BusinessException(EnumClientException.FAILED_TO_UPLOAD_CHUNK_FILE);
            }
        });

        return md5;
    }

    @Override
    public List<BucketVO> files() {

        String metaServerAddress = clientConfig.getMetaServerAddress();


        //  调用meta服务
        Object response = restTemplate.getForObject(
                metaServerAddress + "/meta/files",
                Object.class
        );

        //  将response的结果类型进行转换
        CommonResponse<List<BucketVO>> commonResponse = mapper.convertValue(response,
                new TypeReference<CommonResponse<List<BucketVO>>>() {
        });

        return commonResponse.getData();
    }

    @Override
    public void delete(String bucketName, String filename) {
        String metaServerAddress = clientConfig.getMetaServerAddress();
        String url = "%s/meta/%s/%s".formatted(metaServerAddress,bucketName,filename);

        restTemplate.delete(
                url
        );
    }


    /**
     * 文件的一个分片是冗余存储的，我要获得的元数据信息不是将所有的副本信息都取出，我需要的分片的属性为：
     * 文件名，分片编号，分片起始偏移位，分片大小，分片是否上传成功的标记位
     * 所以一个分片只需拿出一个副本对应的属性即可
     * @param metaFile
     * @return
     */
    private MetaFileVO buildMetaFileVO(MetaFile metaFile){

        MetaFileVO metaFileVO = new MetaFileVO();

        //  将chunks先全部取出
        List<FileChunkMeta> originChunks = metaFile.getChunks();

        //  将chunks做一下筛选，一个chunk只筛出一个副本，填到metafile中
        List<FileChunkMetaVO> fileChunkMetaVOS = new ArrayList<>();

        //  先全部放入
        for (FileChunkMeta originChunk : originChunks) {
            FileChunkMetaVO fileChunkMetaVO = new FileChunkMetaVO();
            fileChunkMetaVO.setFilename(originChunk.getFilename())
                    .setChunkNo(originChunk.getChunkNo())
                    .setChunkStart(originChunk.getChunkStart().intValue())
                    .setChunkSize(originChunk.getChunkSize())
                    .setCompleted(originChunk.getCompleted());

            fileChunkMetaVOS.add(fileChunkMetaVO);
        }

        //  再将全集的重复部分筛掉
        fileChunkMetaVOS = fileChunkMetaVOS.stream().distinct().collect(Collectors.toList());

        return metaFileVO.setChunks(fileChunkMetaVOS)
                .setFilename(metaFileVO.getFilename())
                .setBucketName(metaFileVO.getBucketName());
    }

    public void uploadChunks(MultipartFile file,MetaFile metaFile) throws IOException, ExecutionException, InterruptedException {

        //  将每个分片的元数据取出
        List<FileChunkMeta> chunks = metaFile.getChunks();

        chunks = chunks.stream().sorted((o1, o2) -> o1.getChunkNo().compareTo(o2.getChunkNo()))
                .collect(Collectors.toList());

        //  若未上传分片列表为空，说明该文件已经上传完成
        if (chunks.size() == 0){
            System.out.println("上传分片列表为空");
            return;
        }

        InputStream inputStream = file.getInputStream();

        //  使用任务列表将分片上传任务并行执行
        //  先将每个分片都当做未上传过，后面再处理逻辑
        CompletableFuture<?>[] tasks = new CompletableFuture[chunks.size()];

        byte[] buffer = new byte[0];
        //  记录上一次上传的分片序号是多少
        int preChunkNo = -1;

        //  若有未上传分片，则遍历列表进行上传
        for (int i = 0; i < tasks.length; i++) {

            FileChunkMeta chunk = chunks.get(i);
            System.out.println(chunk.getAddress());

            //  获取一个分片的起始位置
            Long chunkStart = chunk.getChunkStart();

            //  每个分片所存储的内容
            Integer chunkSize = chunk.getChunkSize();

            if (chunk.getChunkNo() != preChunkNo){
                preChunkNo = chunk.getChunkNo();
                //  要重新读取
                buffer = new byte[chunkSize];
                //  只有到这里需要阻塞，其他都可以并行执行
                //  缓冲流不支持偏移读取
                inputStream.read(buffer);
            }

            //  创建异步任务
            byte[] finalBuffer = buffer;
            tasks[i] = CompletableFuture.runAsync(()->{

                //  如果已上传完成，就无需再上传
                if (chunk.getCompleted()){
                    return;
                }

                String md5 = Md5Util.getMd5(finalBuffer);

                FileChunkBO fileChunkBO = new FileChunkBO()
                        .setFilename(chunk.getFilename())
                        .setExtension(chunk.getExtension())
                        .setChunkNo(chunk.getChunkNo())
                        .setChunkSize(chunk.getChunkSize())
                        .setBucketName(metaFile.getBucketName())
                        .setBytes(finalBuffer);

                String schema = chunk.getSchema();

                String address = chunkAddressStrategy.get(chunk);
                Object response = restTemplate.postForObject(address + "/file/write",
                        fileChunkBO,
                        Object.class);
                System.out.println("response = " + response);
                //  为空则上传失败
                if (Objects.isNull(response)){
                    throw new BusinessException("第"+chunk.getChunkNo()+"个分片上传失败",
                            EnumClientException.FAILED_TO_UPLOAD_CHUNK_FILE);
                }

                CommonResponse<String> md5Reponse = mapper.convertValue(response, new TypeReference<CommonResponse<String>>() {
                });

                //  比较Md5是否一致
                if (!md5Reponse.getData().equals(md5)){
                    throw new BusinessException(EnumClientException.CHUNK_FILE_INCOMPLETE);
                }

                //  上传成功则要将completed字段修改掉
                CompleteChunkFileBO completeChunkFileBO = new CompleteChunkFileBO().setFilename(chunk.getFilename())
                        .setChunkNo(chunk.getChunkNo())
                        .setAddress(chunk.getAddress())
                        .setSchema(chunk.getSchema())
                        .setMd5(md5);

                //  将分片信息发送给Meta服务器，让Meta服务器判断该分片是否上传成功，若成功则修改completed字段
                Object resp = restTemplate.postForObject(
                        clientConfig.getMetaServerAddress() + "/meta/chunk/complete",
                        completeChunkFileBO,
                        Object.class
                );
                //  为空则说明分片状态更新失败
                if (Objects.isNull(resp)){
                    throw new BusinessException(EnumClientException.FAILED_TO_UPDATE_CHUNK_FILE_COMPLETED_STATUS);
                }
                log.info("第"+chunk.getChunkNo()+"个分片上传成功，分片状态为："+resp);
            }).whenComplete((o,thorwable)->{
                if (Objects.nonNull(thorwable)){
                    throw new RuntimeException(MessageFormat.format("第{}个分片上传失败",chunk.getChunkNo()));
                }
            });
        }

        //  执行异步任务
        CompletableFuture<Void> allOf = CompletableFuture.allOf(tasks);

        //  要队列里面的任务全部执行完成了才算上传成功
        CompletableFuture<?> anyException = new CompletableFuture<>();
        Arrays.stream(tasks).forEach(t->{
            //  如果有任意一个任务出现异常，就将anyException处理掉
            t.exceptionally(throwable -> {
                anyException.completeExceptionally(throwable);
                return null;
            });
        });
        //  allOf和anyException有一个出现异常则马上停掉所有任务
        CompletableFuture.anyOf(allOf,anyException).get();
    }

}